<?php

namespace RabbitMq;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// RabbitMq 模型
class RabbitMq
{
    static private $instance;
    static private $connection;
    static private $channel;
    const DIRECT = 'direct';
    const TOPIC = 'topic';
    const HEADERS = 'headers';
    const FANOUT = 'fanout';
    static private $exchangeNames = [
        self::DIRECT => 'direct_exchange',
        self::TOPIC => 'topic_exchange',
        self::HEADERS => 'headers_exchange',
        self::FANOUT => 'fanout_exchange',
    ];
    const SEVERITYS = [
        'info',
        'warning',
        'error'
    ];
    static private $exchangeName = '';

    /**
     * RabbitMq constructor.
     * @param $exchangeType
     */
    private function __construct($exchangeType)
    {
        // 取默认配置
        $config = [
            'host' => config('rabbit_mq_conf.host'), //ip
            'port' => config('rabbit_mq_conf.port'),      //端口号
            'user' => config('rabbit_mq_conf.username'),     //用户
            'password' => config('rabbit_mq_conf.password'), //密码
            'vhost' => config('rabbit_mq_conf.vhost')         //虚拟host
        ];

        //创建链接
        self::$connection = new AMQPStreamConnection($config['host'], $config['port'], $config['user'], $config['password'], $config['vhost']);
        self::$channel = self::$connection->channel();
        if (!empty($exchangeType)) {
            self::$exchangeName = self::$exchangeNames[$exchangeType];
            self::$channel->exchange_declare(
                self::$exchangeName, //交换机名称
                $exchangeType, //路由类型
                false, //don't check if a queue with the same name exists 是否检测同名队列
                true, //the queue will not survive server restarts 是否开启队列持久化
                false //the queue will be deleted once the channel is closed. 通道关闭后是否删除队列
            );
        }
    }

    /**
     * 实例化
     * @param string $exchangeType
     * @return RabbitMq
     */
    public static function instance($exchangeType = '')
    {
        if (!self::$instance instanceof self) {
            self::$instance = new self($exchangeType);
        }
        return self::$instance;
    }

    /**
     * 防止被外部复制
     */
    private function __clone()
    {
    }

    /**
     * 简单的发送
     * @param $body
     */
    public function send($body)
    {
        self::$channel->queue_declare('hello', false, false, false);
        $msg = new AMQPMessage($body);
        self::$channel->basic_publish($msg, '', 'hello');
        echo "[X] Sent " . $body . "'\n";
    }

    /**
     * 简单的接收
     * @param $queueName
     * @param $callback
     */
    public function receive($callback)
    {
        self::$channel->queue_declare('hello', false, false, false, true);
        echo "[*] Waiting for messages. To exit press CTRL+C\n";

        self::$channel->basic_consume("hello", "", false, false, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 添加工作队列
     * @param string $data
     * @param string $queueName
     */
    public function addTask($data = "", $queueName)
    {
        self::$channel->queue_declare($queueName, false, true, false, true);
        if (empty($data)) {
            $data = 'Hello World!';
        }
        $msg = new AMQPMessage(
            $data,
            array("delivery_mode" => AMQPMessage::DELIVERY_MODE_PERSISTENT)
        );
        self::$channel->basic_publish($msg, "", $queueName);

//        echo "[x] Sent $data \n";
    }

    /**
     * 执行工作队列
     * @param $callback
     * @param $queueName
     */
    public function workTask($callback, $queueName)
    {
        self::$channel->queue_declare($queueName, false, true, false, true);
        echo " [*] Waiting for messages. To exit press CTRL+C", "\n";

        self::$channel->basic_qos(null, 1, null);
        self::$channel->basic_consume($queueName, "", false, false, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 发布
     * @param string $data
     */
    public function sendQueue($data = "")
    {
        if (empty($data)) {
            $data = 'info:Hello World!';
        }
        $msg = new AMQPMessage($data);
        self::$channel->basic_publish($msg, self::$exchangeName);
        echo "[x] Sent $data \n";
    }

    /**
     * 订阅
     * @param $callback
     */
    public function subscribeQueue($callback)
    {
        list($queue_name, ,) = self::$channel->queue_declare(
            "", //队列名称
            false, //dont check if a queue with the same name exists 是否检测同名队列
            true, //the queue will not survive server restarts 是否开启队列持久化
            true, //the queue might be accessed by other channels 队列是否可以被其他队列访问
            false //the queue will be deleted once the channel is closed. 通道关闭后是否删除队列
        );
        self::$channel->queue_bind($queue_name, self::$exchangeName);
        echo "[*] Waiting for logs. To exit press CTRL+C \n";
        self::$channel->basic_consume($queue_name, "", false, true, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 发送（直接交换机）
     * @param $routingKey
     * @param string $data
     */
    public function sendDirect($routingKey, $data = "")
    {
        if (empty($data)) {
            $data = "Hello World!";
        }
        $msg = new AMQPMessage($data);
        self::$channel->basic_publish($msg, self::$exchangeName, $routingKey);
        echo "[x] Sent $routingKey:$data \n";
    }

    /**
     * 接收（直接交换机）
     * @param \Closure $callback
     * @param array $bindingKeys
     */
    public function receiveDirect(\Closure $callback, array $bindingKeys)
    {
        list($queue_namme, ,) = self::$channel->queue_declare("", false, true, true, false);
        foreach ($bindingKeys as $bindingKey) {
            self::$channel->queue_bind($queue_namme, self::$exchangeName, $bindingKey);
        }
        echo "[x] Waiting for logs. To exit press CTRL+C \n";
        self::$channel->basic_consume($queue_namme, "", false, true, false, false, $callback);
        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 发送（主题交换机）
     * @param $routingKey
     * @param string $data
     */
    public function sendTopic($routingKey, $data = "")
    {
        if (empty($data)) {
            $data = "Hello World!";
        }
        $msg = new AMQPMessage($data);
        self::$channel->basic_publish($msg, self::$exchangeName, $routingKey);
        echo " [x] Sent ", $routingKey, ':', $data, " \n";
    }

    /**
     * 接收（主题交换机）
     * @param \Closure $callback
     * @param array $bindingKeys
     */
    public function receiveTopic(\Closure $callback, array $bindingKeys)
    {
        list($queueName, ,) = self::$channel->queue_declare("", false, true, true, false);
        foreach ($bindingKeys as $bindingKey) {
            self::$channel->queue_bind($queueName, self::$exchangeName, $bindingKey);
        }

        echo ' [*] Waiting for logs. To exit press CTRL+C', "\n";
        self::$channel->basic_consume($queueName, '', false, true, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 销毁
     */
    public function __destruct()
    {
        // TODO: Implement __destruct() method.
        self::$channel->close();
        self::$connection->close();
    }
}
